package com.coincalf.task.task;

import com.coincalf.feign.MarketServiceClient;
import com.coincalf.framework.constants.Constant;
import com.coincalf.framework.enums.BaseStatus;
import com.coincalf.task.event.KlineEvent;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author qiang
 */
@Slf4j
@Component
public class TopicKLineTask implements Constant {

    @Value("${subscribe.enable}")
    private boolean subscribeEnable;

    @Autowired
    private MarketServiceClient marketServiceClient;

    /**
     * 每3秒推送K线数据
     */
    @Scheduled(fixedRate = 3000)
    public void pushKline() {
        if (subscribeEnable) {
            ExecutorService executor = new ThreadPoolExecutor(
                    5,
                    10,
                    100L,
                    TimeUnit.MILLISECONDS,
                    new LinkedBlockingQueue<>(30),
                    new ThreadPoolExecutor.CallerRunsPolicy()
            );
            String channel = "market.%s.kline.%s";
            marketServiceClient.tradeMarkets().forEach(market -> {
                if (market.getStatus() == BaseStatus.EFFECTIVE.getCode()) {
                    executor.submit(new KlineEvent(market.getSymbol(), channel, REDIS_KEY_TRADE_KLINE));
                }
            });
            executor.shutdown();
        }
    }

    /**
     * 每3秒推送K线数据
     */
    @Scheduled(fixedRate = 3000)
    public void pushForexKline() {
        if (subscribeEnable) {
            ExecutorService executor = new ThreadPoolExecutor(
                    5,
                    10,
                    100L, TimeUnit.MILLISECONDS,
                    new LinkedBlockingQueue<>(30),
                    new ThreadPoolExecutor.CallerRunsPolicy()
            );
            String channel = "market.tc.%s.kline.%s";
            marketServiceClient.forexMarkets().forEach(market -> {
                if (market.getStatus() == BaseStatus.EFFECTIVE.getCode()) {
                    executor.submit(new KlineEvent(market.getSymbol(), channel, REDIS_KEY_FOREX_KLINE));
                }
            });
            executor.shutdown();
        }
    }

}
